package com.heima.behavior.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.behavior.entity.ApLikesBehavior;
import com.heima.behavior.mapper.ApLikesBehaviorMapper;
import com.heima.behavior.service.ApLikesBehaviorService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.common.threadlocal.UserThreadLocalUtils;
import com.heima.common.util.JsonUtils;
import com.heima.model.article.dtos.ArticleBehaviorMsg;
import com.heima.model.article.dtos.UserBehaviorMsgDto;
import com.heima.model.behavior.dtos.LikesBehaviorDto;
import com.heima.model.user.dtos.UserRelationDto;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Objects;

/**
 * <p>
 * APP点赞行为表 服务实现类
 * </p>
 *
 * @author HM
 * @since 2023-04-09
 */
@Service
public class ApLikesBehaviorServiceImpl extends ServiceImpl<ApLikesBehaviorMapper, ApLikesBehavior> implements ApLikesBehaviorService {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private KafkaProducer kafkaProducer;

    @Override
    public void likesBehavior(LikesBehaviorDto dto) {
        //获取当前用户id
        Integer userId = UserThreadLocalUtils.getUserId();
        //从redis中读取数据
        String json = (String) redisTemplate.opsForHash().get("behavior:" + userId, dto.getArticleId());
        UserBehaviorMsgDto userBehaviorMsgDto = JsonUtils.toBean(json, UserBehaviorMsgDto.class);
        //判断redis中是否存在，没有则new一个
        if (Objects.isNull(userBehaviorMsgDto)) {
            userBehaviorMsgDto = new UserBehaviorMsgDto();
        }

        //查找数据库中点赞的对象
        LambdaQueryWrapper<ApLikesBehavior> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ApLikesBehavior::getApUserId, userId);
        queryWrapper.eq(ApLikesBehavior::getArticleId, dto.getArticleId());
        ApLikesBehavior apLikesBehaviorDB = this.getOne(queryWrapper);
        //判断数据库中是否存在
        if (Objects.isNull(apLikesBehaviorDB)) {
            //不存在则新建  新建都是点赞

            //封装数据
            ApLikesBehavior apLikesBehavior = createApLikesBehavior(dto, userId);
            //存数据库
            this.save(apLikesBehavior);
            //设置成点赞
            userBehaviorMsgDto.setIslike(true);
            //发送kafka更新热点 点赞 +1
            sendKafka(userBehaviorMsgDto, dto);
        } else {
            //存在则修改  包含点赞和取消点赞

            //封装数据
            apLikesBehaviorDB.setOperation(dto.getOperation()==1?true:false);
            apLikesBehaviorDB.setUpdateTime(new Date());
            //更新数据库
            this.updateById(apLikesBehaviorDB);
            //设置点赞状态
            userBehaviorMsgDto.setIslike(dto.getOperation()==1?true:false);
            //发送kafka更新热点
            sendKafka(userBehaviorMsgDto, dto);
        }
        //发送到redis
        redisTemplate.opsForHash().put("behavior:" + userId, dto.getArticleId(), JsonUtils.toString(userBehaviorMsgDto));
    }

    private static ApLikesBehavior createApLikesBehavior(LikesBehaviorDto dto, Integer userId) {
        ApLikesBehavior apLikesBehavior = new ApLikesBehavior();
        apLikesBehavior.setArticleId(dto.getArticleId());
        apLikesBehavior.setApUserId(userId);
        apLikesBehavior.setEquipmentId(dto.getEquipmentId());
        apLikesBehavior.setType(dto.getType());
        apLikesBehavior.setOperation(dto.getOperation()==1?true:false);
        apLikesBehavior.setCreatedTime(new Date());
        apLikesBehavior.setUpdateTime(new Date());
        return apLikesBehavior;
    }

    public void sendKafka(UserBehaviorMsgDto userBehaviorMsgDto, LikesBehaviorDto likesBehaviorDto) {
        ArticleBehaviorMsg msg = new ArticleBehaviorMsg();
        msg.setArticleId(likesBehaviorDto.getArticleId());
        //点赞+1  取消点赞—1
        msg.setNum(userBehaviorMsgDto.getIslike() == true ? 1 : -1);
        msg.setType(ArticleBehaviorMsg.ArticleBehaviorType.LIKES);
        ProducerRecord<String, String> record1 = new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JsonUtils.toString(msg));
        kafkaProducer.send(record1);
        System.out.println("发送数据" + JsonUtils.toString(msg));
    }
}
